package cn.dmp.charts.media



import java.sql.DriverManager

import cn.dmp.utils.ParseUtils
import com.mysql.jdbc.{Connection, PreparedStatement}
import org.apache.commons.lang.StringUtils
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD


object MediaAnalyzeRptV2 {
  def main(args: Array[String]): Unit = {
    if (args.length != 3) {
      println(
        """
          |cn.dmp.report.ProCityRpt
          |参数：
          | logInputPath
          | appMappingtxtInputPath
          | outputPath
        """.stripMargin)
      sys.exit()
    }

    val Array(logInputPath,appMappingtxtInputPath,outputPath) = args

    val conf: SparkConf = new SparkConf().setAppName("ArealDistribution171106V3").setMaster("local[4]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.sql.parquet.compression.codec", "snappy")

    val sc: SparkContext = new SparkContext(conf)

    //广播appid匹配规则
    //(appId,appName)
    //字典文件
    val dictMap = sc.textFile(appMappingtxtInputPath).map(_.split("\t",-1)).map(t => {
      (t(4), t(1))
    }).collect().toMap
    //发送广播出去
   val dictRef: Broadcast[Map[String, String]] = sc.broadcast(dictMap)


    val rawRDD: RDD[String] = sc.textFile(logInputPath)

    val splitsRDD: RDD[Array[String]] = rawRDD.map(_.split(",",-1))

    //过滤
    val suitedLengthRDD = splitsRDD.filter(_.length >= 85)

    //取出判断条件字段
    val baseRDD = suitedLengthRDD.map(splits => {


      val requestmode: Int = ParseUtils.parseInt(splits(8))
      val processnode: Int = ParseUtils.parseInt(splits(35))
      val iseffective: Int = ParseUtils.parseInt(splits(30))
      val isbilling: Int = ParseUtils.parseInt(splits(31))
      val isbid: Int = ParseUtils.parseInt(splits(39))
      val iswin: Int = ParseUtils.parseInt(splits(42))
      val adorderid: Int = ParseUtils.parseInt(splits(2))
      val winprice: Double = ParseUtils.parseDouble(splits(41))
      val adpayment: Double = ParseUtils.parseDouble(splits(75))

      val appid: String = splits(13)
      val appname: String = splits(14)
//t._10 = appname  t._11 = appid



      (requestmode, processnode, iseffective, isbilling, isbid, iswin, adorderid, winprice, adpayment,appname,appid)

    })
    //匹配条件
    val adMatchingRDD = baseRDD.map(f = t => {

      val primaryRequest = if (t._1 == 1 && t._2 >= 1) 1 else 0

      val effectiveRequest = if (t._1 == 1 && t._2 >= 2) 1 else 0

      val AdRequest = if (t._1 == 1 && t._2 == 3) 1 else 0

      val biddingTimes = if (t._3 == 1 && t._4 == 1 && t._5 == 1 && t._7 != 0) 1 else 0

      val succesedBiddingTime = if (t._3 == 1 && t._4 == 1 && t._6 == 1) 1 else 0

      val showTimes = if (t._1 == 2 && t._3 == 1) 1 else 0

      val clickTimes = if (t._1 == 3 && t._3 == 1) 1 else 0

      val AdConsume = if (t._3 == 1 && t._4 == 1 && t._6 == 1) 1.0 * t._8 / 1000 else 0.0

      val AdCost = if (t._3 == 1 && t._4 == 1 && t._6 == 1) 1.0 * t._9 / 1000 else 0.0


      var newAppName = t._10
      if (!StringUtils.isNotEmpty(newAppName)) {
        newAppName = dictRef.value.getOrElse(t._11, "未知")
      }

      (newAppName, (primaryRequest, effectiveRequest, AdRequest, biddingTimes, succesedBiddingTime, showTimes, clickTimes, AdConsume, AdCost))
    })

    val result: RDD[(String, (Int, Int, Int, Int, Int, Int, Int, Double, Double))] = adMatchingRDD.reduceByKey((t1, t2) => {
      ((t1._1 + t2._1), (t1._2 + t2._2), (t1._3 + t2._3), (t1._4 + t2._4), (t1._5 + t2._5), (t1._6 + t2._6), (t1._7 + t2._7), (t1._8 + t2._8), (t1._9 + t2._9))
    })

    result.map(t =>t._1 +"," +t._2._1+","+t._2._2+","+t._2._3+","+t._2._4+","+t._2._5+","+t._2._6+","+t._2._7+","+t._2._8+","+t._2._9 )
      .saveAsTextFile(outputPath)


    sc.stop()
     // (appname, (primaryRequest, effectiveRequest, AdRequest, biddingTimes, succesedBiddingTime, showTimes, clickTimes, AdConsume, AdCost))

//    })
//
//    val reducedRDD = adMatchingRDD.reduceByKey((t1, t2) => {
 //     ((t1._1 + t2._1), (t1._2 + t2._2), (t1._3 + t2._3), (t1._4 + t2._4), (t1._5 + t2._5), (t1._6 + t2._6), (t1._7 + t2._7), (t1._8 + t2._8), (t1._9 + t2._9))
//    })


      //打印测试
//    reducedRDD.foreach(x=>{
//      println(x)
//    })







  }

}
